﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

#region

using System;
using System.Diagnostics;
using System.IO;
using AzureMRCore;
using AzureMRCore.DataModel;
using AzureMRCore.DataTypes;
using AzureMRCore.Drivers;
using AzureMRCore.InputFormat;
using AzureMRCore.MapRed;
using AzureMRCore.OutputFormat;
using Ionic.Zip;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;

#endregion

namespace BlastWorkerSample
{
    public class BlastMR : MapOnlyDriver<StringKey, StringValue, StringKey, FileValue>
    {
        public override string Name
        {
            get { return "BlastMR"; }
        }

        public override Mapper<StringKey, StringValue, StringKey, FileValue, NullKey, NullValue> MapperType
        {
            get { return new BlastMapper(); }
        }

        public override IInputFormat<StringKey, StringValue> InputFormatType
        {
            get { return new FileInputFormat(); }
        }

        public override IOutputFormat<StringKey, FileValue> MapOutputFormatType
        {
            get { return new FileOutputFormat<StringKey>(); }
        }

        public override Boolean ConfigMapRed(MRConf mrConf, CloudStorageAccount storageAccount, out object configOut)
        {
            configOut = DownloadDB(storageAccount);
            //configOut = "C:\\Users\\thilina\\Desktop\\temp\\blastdb\\db\\nr";
            return true;
        }

        protected string DownloadDB(CloudStorageAccount storageAccount)
        {
            try
            {
                string dbPath;
                const string blobURI = "http://twisterazure.blob.core.windows.net/blastdb/db.zip";
                const string localFileName = "db.zip";
                const string dirName = "blastdb";
                Log("Start downloading Blast DB.", storageAccount);
                Stopwatch sw = new Stopwatch();
                sw.Start();
                DirectoryInfo dir = DownloadAndExtract(storageAccount, blobURI, localFileName, dirName);
                dbPath = dir.FullName + "\\db\\nr";
                //_dbPath = "C:\\Users\\thilina\\Desktop\\temp\\blastdb\\db\\nr";

                sw.Stop();
                string message = "BlastDownload dbPath=" + dbPath + " time:" + sw.ElapsedMilliseconds;
                Trace.WriteLine(message);

                Log(message, storageAccount);
                return dbPath;
            }
            catch (Exception e)
            {
                Trace.TraceError(e.ToString());
                CloudTableClient tableClient = storageAccount.CreateCloudTableClient();
                var errorContext = new ErrorDataContext(storageAccount.TableEndpoint.ToString(),
                                                        storageAccount.Credentials, "AzureMRErrors");
                tableClient.CreateTableIfNotExist(errorContext.Name);
                var error = new ErrorDataModel(e, "maptask", e.Message);
                errorContext.AddObject(errorContext.Name, error);
                errorContext.SaveChangesWithRetries();
            }
            return null;
        }

        public DirectoryInfo DownloadAndExtract(CloudStorageAccount storageAccount, string blobURI,
                                                string localFileName, string dirName)
        {
            CloudBlobClient blobClient = storageAccount.CreateCloudBlobClient();
            blobClient.ParallelOperationThreadCount = 10;

            LocalResource localResource = RoleEnvironment.GetLocalResource("temp");

            CloudBlob fileBlob = blobClient.GetBlobReference(blobURI);

            DirectoryInfo dir = Directory.CreateDirectory(localResource.RootPath + "\\" + dirName);
            string localInputFileName = localResource.RootPath + "\\" + localFileName;

            var fileinfo = new FileInfo(localInputFileName);
            if (!fileinfo.Exists)
            {
                Trace.WriteLine("Downloading Blast DB");
                Log("Downloading Blast DB", storageAccount);
                BlobRequestOptions options = new BlobRequestOptions();
                options.Timeout = new TimeSpan(0, 4, 0, 0);
                //fileBlob.DownloadToFile(localInputFileName, options);
                using (FileStream fs = new FileStream(localInputFileName, FileMode.Create))
                {
                    // I’m using an four-hour timeout
                    fileBlob.DownloadToStream(fs, new BlobRequestOptions {Timeout = new TimeSpan(4, 0, 0)});
                }
            }
            else
            {
                Log("File exists", storageAccount);
                Trace.WriteLine("File exists");
            }

            Log("Start Extracting the DB", storageAccount);
            Trace.WriteLine("Start Extracting the DB");

            using (ZipFile zip1 = ZipFile.Read(localInputFileName))
            {
                // here, we extract every entry, but we could extract conditionally
                // based on entry name, size, date, checkbox status, etc.  
                foreach (ZipEntry e in zip1)
                {
                    e.Extract(dir.FullName, ExtractExistingFileAction.DoNotOverwrite);
                }
            }
            return dir;
        }

        private void Log(string message, CloudStorageAccount storageAccount)
        {
            CloudTableClient tableClient = storageAccount.CreateCloudTableClient();
            ErrorDataContext errorContext = new ErrorDataContext(storageAccount.TableEndpoint.ToString(),
                                                                 storageAccount.Credentials, "AzureMRErrors");
            tableClient.CreateTableIfNotExist(errorContext.Name);
            ErrorDataModel error =
                new ErrorDataModel(RoleEnvironment.CurrentRoleInstance.Id + "_  " + message);
            errorContext.AddObject(errorContext.Name, error);
            errorContext.SaveChangesWithRetries();
        }
    }
}